package com.yeming.flink.practice.source

import org.apache.flink.streaming.api.scala._

/**
 * 基站日志
 *
 * @param sid 基站ID
 * @param callOut 主叫号码
 * @param callIn 被叫号码
 * @param callType 呼叫类型
 * @param callTime 呼叫时间（毫秒）
 * @param duration 呼叫时长（秒）
 */
case class StationLog(sid:String, var callOut:String, var callIn:String, callType:String, callTime:Long, duration:Long)

object CollectionSource {

  def main(args: Array[String]): Unit = {
    //初始化环境变量
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    streamEnv.setParallelism(1)
    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.fromCollection(Array(
      new StationLog("001", "132", "185", "busy", System.currentTimeMillis(), 0),
      new StationLog("002", "142", "195", "busy", System.currentTimeMillis(), 0),
      new StationLog("003", "139", "155", "busy", System.currentTimeMillis(), 0),
      new StationLog("004", "185", "132", "success", System.currentTimeMillis(), 20),
    ))
    //Transformation
    //Sink
    stream.print()
    //执行
    streamEnv.execute("CollectionSourceDemo")
  }

}
